package com.taldh.netty.learning.multiReactorMultiThread;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * date 2020/8/11
 *
 * @author 袁小黑
 * @version 1.0.0
 **/

public class ReactorMm implements Runnable{

    public ServerSocketChannel serverSocket;
    public Selector selector;

    public Selector[] selectors = new Selector[Runtime.getRuntime().availableProcessors() * 2]; //sub reactor

    ReactorMm(SocketAddress local) throws IOException {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        this.serverSocket.bind(local);
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //注册accept事件
        sk.attach(new Acceptor()); //调用Acceptor()为回调方法
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                if (selector.select()>0) {
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey selectionKey : selectionKeys) {
                        dispatchKey(selectionKey);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
            }
        }
    }

    private void dispatchKey(SelectionKey selectionKey) {
        Object r = selectionKey.attachment(); //调用SelectionKey绑定的调用对象
        if (r instanceof Runnable)
            ((Runnable)r).run();
    }

    class Acceptor implements Runnable {
        int next = 0;
        @Override
        public synchronized void run() {
            //...
            SocketChannel connection = null;
            try {
                connection = serverSocket.accept();
            } catch (IOException e) {
                //
            }
            if (connection != null) {
                try {
                    new MultiThreadEchoHandler(selectors[next], connection);
                } catch (IOException e) {
                    //
                }
            }
            if (++next == selectors.length)
                next = 0;
        }
    }

    public static void main(String[] args) throws IOException {
        new Thread(new ReactorMm(new InetSocketAddress("127.0.0.1", 8080)));
    }
}


class Handler {
    final SocketChannel socket;
    final SelectionKey sk;
    ByteBuffer input = ByteBuffer.allocate(1024);
    ByteBuffer output = ByteBuffer.allocate(1024);
    static final int READING = 0, SENDING = 1, PROCESSING= 3;
    int state = READING;
    ExecutorService pool = Executors.newFixedThreadPool(2 * Runtime.getRuntime().availableProcessors()); //声明线程池

    Handler(Selector sel, SocketChannel c) throws IOException {
        socket = c;
        c.configureBlocking(false);
        // Optionally try first read now
        sk = socket.register(sel, 0);
        sk.attach(this); //将Handler绑定到SelectionKey上
        sk.interestOps(SelectionKey.OP_READ);
        sel.wakeup();
    }
    boolean inputIsComplete() {
        /* ... 半包粘包的问题 */
        return false;
    }
    boolean outputIsComplete() {
        /* ...半包粘包的问题 */
        return false;
    }
    void process() { /* ... */ }

    public void run() throws IOException {
        socket.read(input);
        if (inputIsComplete()) {
            process();
            sk.attach(new Sender());
            sk.interestOps(SelectionKey.OP_WRITE);
            sk.selector().wakeup();
        }
    }

    class Sender implements Runnable {
        public void run(){ // ...
            try {
                socket.write(output);
            } catch (IOException e) {
                //
            }
            if (outputIsComplete()) sk.cancel();
        }
    }

    synchronized void read() throws IOException {
        socket.read(input);
        if (inputIsComplete()) {
            state = PROCESSING;
            pool.execute(new Processer());//处理程序放在线程池中执行
        }
    }
    void send() throws IOException {
        socket.write(output);
        if (outputIsComplete()) sk.cancel();
    }

    synchronized void processAndHandOff() {
        process();
        state = SENDING; // or rebind attachment
        sk.interestOps(SelectionKey.OP_WRITE);
    }

    class Processer implements Runnable {
        public void run() {
            processAndHandOff();
        }
    }
}